package hbase.mapreduce;

import org.apache.commons.codec.digest.DigestUtils;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Mutation;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

/**
 * Created by spark on 16-10-23.
 */
public class ImportFromFile extends Configured implements Tool{


    public static final String NAME = "ImportFromFile";

    public enum Counters { LINE }

    @Override
    public int run(String[] args) throws Exception {

        if (args.length != 3) {
            System.err.println("Usage: ImportFromFile tableName inputfile family:qualifier");
            GenericOptionsParser.printGenericCommandUsage(System.err);
            return 1;
        }

        getConf().setStrings("conf.column", args[2]);

//        getConf().addResource("hadoop-master.xml");

        Job job = Job.getInstance(getConf(), "Import from file " + args[1] + " to table " + args[0]);
        job.setJar("./target/hadoop-test-1.0-SNAPSHOT.jar");
        TableMapReduceUtil.addHBaseDependencyJars(getConf());

        job.setMapperClass(ImportMapper.class);
        job.setJarByClass(ImportFromFile.class);
        job.setOutputFormatClass(TableOutputFormat.class);
        job.getConfiguration().setStrings(TableOutputFormat.OUTPUT_TABLE, args[0]);
        job.setOutputKeyClass(ImmutableBytesWritable.class);
        job.setOutputValueClass(Mutation.class);

        job.setNumReduceTasks(0);

        FileInputFormat.addInputPath(job, new Path(args[1]));


        return job.waitForCompletion(true) ? 0 : 1;
    }

    public static class ImportMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, Mutation> {

        private byte[] family = null;
        private byte[] qualifier = null;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            String column = context.getConfiguration().get("conf.column");
            byte[][] byteColumn = KeyValue.parseColumn(Bytes.toBytes(column));
            family = byteColumn[0];
            if (byteColumn.length > 1) {
                qualifier = byteColumn[1];
            }
        }

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

            byte[] row = DigestUtils.md5(value.toString());
            Put put = new Put(row);
            put.addColumn(family, qualifier, Bytes.toBytes(value.toString()));
            context.write(new ImmutableBytesWritable(row), put);
            context.getCounter(Counters.LINE).increment(1);
        }
    }


    public static void main(String[] args) throws Exception {
        int exitCode = ToolRunner.run(new ImportFromFile(), args);
        System.exit(exitCode);
    }

}
